package verticle;

import io.vertx.core.file.OpenOptions;
import io.vertx.reactivex.core.AbstractVerticle;
import io.vertx.reactivex.core.buffer.Buffer;
import io.vertx.reactivex.core.parsetools.RecordParser;

import java.io.FileNotFoundException;
import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;

/**
 * verticle.SocketVerticle
 *
 * @author liyixin
 * @date 2018/12/24
 */
public class SocketVerticle extends AbstractVerticle {

    @Override
    public void start() {
        vertx.createNetServer().connectHandler(sock -> {

            RecordParser parser = RecordParser.newDelimited("\n", sock);
            parser.handler(buf -> {
                int size = Integer.parseInt(buf.toJsonObject().getJsonObject("dataMap").getString("obj_size"));
                int size1 = size + 1024;
                vertx.fileSystem().open("/fs-SP2-1/test/test", new OpenOptions(),file -> {
                    sock.toFlowable()
                            .doOnNext(file.result()::write)
                            .map(Buffer::length)
                            .reduce((x, y) -> {
                                int sum = x + y;
                                if (sum >= size) {
                                    sock.write("B89117B0DD1FDC9863202A8E25FDFC83*").close();
                                }
                                return sum;
                            })
                            .subscribe();
                    sock.write("ok*");
                });
            });
        }).listen(11121);
    }
}
